package ckR

import (
	"db2s/global"
)

//func (context *SchedulePlanContext) Check_bak() {
//	var wg sync.WaitGroup
//	fmt.Println(fmt.Sprintf("-- [%v] %v report: start schedule task monitor -- ", time.Now().Format("2006-01-02 15:04:05"), global.ToolName))
//	terminal(&wg, context)
//	context.monitor()
//	wg.Add(1)
//	if !chanNil(ResultStatisticsMq) {
//		go resultAsyncActive(&wg)
//	} else {
//		wg.Done()
//	}
//	closeObject.AbnormalExit()
//	//执行计划开始
//	schedulePlanStart()
//	global.MyWait.Wait()
//	//关闭检测
//	context.Close()
//	//关闭消费者 阻塞
//	consumer.Shutdown()
//	//关闭生产者
//	producer.Shutdown()
//	if !chanNil(BarResultStatisticMq) {
//		close(BarResultStatisticMq)
//	}
//	newTer.Stop()
//	wg.Wait()
//	closeObject.CloseAction()
//	fmt.Println("")
//	//输出结果信息
//	fmt.Println("Check end")
//	infoWriteInfo(fmt.Sprintf("%v check object {%v} complete !!!", global.ToolName, GlobalPConfigs.rules.CheckMode))
//	fmt.Println(fmt.Sprintf("** %v Overview of results **", global.ToolName))
//	fmt.Println("Check time: ", fmt.Sprintf("%.2fs", time.Since(context.BeginTime).Seconds()), "(Seconds)")
//	newTer.ExecTime = fmt.Sprintf("%v", time.Since(context.BeginTime).Seconds())
//	newTer.Print()
//}

// sendStart
//func schedulePlanStart() {
//	err := PlanContext.Pool.Submit(func() {
//		fmt.Println(fmt.Sprintf("-- [%v] %v report: start schedule task check -- ", time.Now().Format("2006-01-02 15:04:05"), global.ToolName))
//		for {
//			select {
//			case k, e := <-PlanContext.ChanSchedulePlanTable:
//				if !e {
//					PlanContext.objectEndMark(getCheckMod())
//					return
//				}
//				if k.TPod.TableMiss {
//					missTableAction(k)
//					continue
//				}
//				global.GTableCurry <- struct{}{}
//				if funcName, ok := FuncRef[GlobalPConfigs.rules.CheckMode]; ok {
//					if funcName.Kind() == reflect.Func {
//						arguments := []reflect.Value{reflect.ValueOf(k)}
//						funcName.Call(arguments)
//					}
//				} else {
//					fmt.Println(fmt.Sprintf("-- [%v] %v report: checkObject parameter selection error, please refer to the log file for details, enable debug to get more information -- ", time.Now().Format("2006-01-02 15:04:05"), global.ToolName))
//					os.Exit(1)
//				}
//			}
//		}
//	})
//	if err != nil {
//		return
//	}
//}

// // InitToSchedulePlan2Table 初始化 执行计划 以 map 的形式 key 表名 value SchedulePlan2
//
//	func InitToSchedulePlan2Table_bak() (ChanSchedulePlanTable chan *SchedulePlan2) {
//		var f2 = global.GetIoc().GetBean("schemaInitContext").(*SchemaInitContext)
//		ChanSchedulePlanTable = make(chan *SchedulePlan2, f2.TableSum)
//		global.MyWait.Add(1)
//		go func() {
//			for {
//				select {
//				case x, y := <-f2.ChanSchemaContext:
//					if !y {
//						return
//					}
//					if x == nil {
//						continue
//					}
//				}
//			}
//		}()
//		return
//	}
//
// speedLimitInit 目标端写入的相关优化
func speedLimitInit(plan2 *SchedulePlan2) {
	//var (
	//	readSum        int64  = 0
	//	writeSum       int64  = 0
	//	readParallel   int64  = 0
	//	writeParallel  int64  = 0
	//	readAvgTime    uint64 = 0
	//	writeAvgTime   uint64 = 0
	//	status         int64  = 0
	//	readExecCount  int64  = 0
	//	writeExecCount int64  = 0
	//)
	//plan2.TableSpeedLimit = global.TableSpeedLimit{
	//	ReadSum:        &readSum,
	//	WriteSum:       &writeSum,
	//	ReadParallel:   &readParallel,
	//	WriteParallel:  &writeParallel,
	//	ReadAvgTime:    &readAvgTime,
	//	WriteAvgTime:   &writeAvgTime,
	//	Status:         &status,
	//	ReadExecCount:  &readExecCount,
	//	WriteExecCount: &writeExecCount,
	//	SpeedLimit:     limit.NewRateLimiter(1000000000, 1),
	//	InstanceConfig: global.DBInstanceConfig{
	//		LowPositionConn:   GlobalPConfigs.InstanceConfig.LowPositionConn,
	//		CurryPositionConn: GlobalPConfigs.InstanceConfig.CurryPositionConn,
	//		HighPositionConn:  GlobalPConfigs.InstanceConfig.HighPositionConn,
	//	},
	//}
}
func productCustomerMsgInit() global.MqProductCustomerSendMsgSum {
	var (
		productMsg               int64 = 0
		customerMsg              int64 = 0
		productObject            int64 = 0
		customerObject           int64 = 0
		productSendMsgFinishMark int64 = 0
	)
	return global.MqProductCustomerSendMsgSum{
		ProductMsg:               &productMsg,
		CustomerMsg:              &customerMsg,
		ProductObject:            &productObject,
		CustomerObject:           &customerObject,
		ProductSendMsgFinishMark: &productSendMsgFinishMark,
	}
}
func tailMessageInit(plan2 *SchedulePlan2) {
	//var (
	//	product    int64 = 0
	//	customer   int64 = 0
	//	mqStart    int64 = 0
	//	mqLevels   int64 = 0
	//	mqLevel    int64 = 0
	//	objectSum  int64 = 0
	//	abT        int64 = 0
	//	intT       int64 = 0
	//	stringT    int64 = 0
	//	timeT      int64 = 0
	//	missT      int64 = 0
	//	redundantT int64 = 0
	//)
	//plan2.mqProductCustomerMonitor = global.MqProductCustomerMonitorMessage{
	//	FirstMsg:               productCustomerMsgInit(),
	//	SecondMsg:              productCustomerMsgInit(),
	//	MqProductSeq:           &product,
	//	MqCustomerSeq:          &customer,
	//	MqStartTiming:          &mqStart,
	//	MqLevelProductSend:     &mqLevels,
	//	MqLevelCustomerReceive: &mqLevel,
	//	ObjectSum:              &objectSum,
	//	ProductMessage:         make(chan map[string][]string, 10000),
	//}
	//plan2.statistics = global.Statistics{
	//	AbTotal:        &abT,
	//	IntTotal:       &intT,
	//	StringTotal:    &stringT,
	//	TimeTotal:      &timeT,
	//	MissTotal:      &missT,
	//	RedundantTotal: &redundantT,
	//}
}

//type chanSendMsg struct {
//	Object           string
//	SendData         chan any
//	SendMsgInfo      SendMsgInfo
//	ExecFunc         func()
//	SpeedLimitSwitch bool
//	SpeedLimit       *limit.RateLimiter
//	Status           *int64
//	SendMsg          sendTopicF
//	FatherSon        FatherSon
//	SendMsgEnd       sendEndActive
//}

// 循环读取管道并发送消息
//func chanMonitorSendMsg(m *chanSendMsg) bool {
//var (
//	uid    uuid.UUID
//	event  = "[chanMonitorSendMsg]"
//	logSeq int64
//)
//pc, _, _, _ := runtime.Caller(1)
//callingFunc := runtime.FuncForPC(pc)
//callFile, callLine := callingFunc.FileLine(pc)
//callFunc := callingFunc.Name()
//callFuncInfo := fmt.Sprintf("Call functions %v Call File %v:%v", callFunc, callFile, callLine)
//debugWriteInfo(fmt.Sprintf("(%d) %v %v Start reading chan and send topic messages.", logSeq, callFuncInfo, event))
//if v, ok := m.FatherSon.FatherTags["logSeq"]; !ok {
//	debugWriteInfo(fmt.Sprintf("(%d) %v %v The logseq value was not obtained and the sending failed.", logSeq, callFuncInfo, event))
//	return false
//} else {
//	logSeq = v.(int64)
//}
//var status = global.IsEmptyStruct(m.SpeedLimit)
//for {
//	if m.SpeedLimitSwitch {
//		if atomic.LoadInt64(m.Status) == 1 {
//			if !status && !m.SpeedLimit.TryAcquire() {
//				continue
//			}
//		}
//	}
//	select {
//	case c, ok := <-m.SendData:
//		if !ok {
//			m.SendMsgEnd()
//			return true
//		} else {
//			if m.ExecFunc != nil {
//				m.ExecFunc()
//			}
//			var ms = make(map[string]any)
//			uid = uuid.New()
//			for k, v := range m.FatherSon.FatherTags {
//				if strings.EqualFold(k, "sendMsgKey") {
//					ms[v.(string)] = c
//				} else {
//					ms[k] = v
//				}
//			}
//			//标签生成：
//			if v1, ok1 := ms["label"]; ok1 { //标签标识
//				for k2 := range v1.(map[string]any) { //循环集合
//					if strings.EqualFold(k2, "no") { //first label
//						ms["label"] = map[string]any{uid.String(): []string{}}
//						m.FatherSon.FatherSonTags[uid.String()] = []string{}
//					} else if len(k2) > 0 { //子级标签
//						if v4, ok4 := m.FatherSon.FatherSonTags[k2]; ok4 {
//							v4 = append(v4, uid.String())
//							m.FatherSon.FatherSonTags[k2] = v4
//						} else {
//							m.FatherSon.FatherSonTags[k2] = []string{uid.String()}
//						}
//						ms["label"] = map[string]any{k2: uid.String()}
//					}
//				}
//			}
//			if !m.SendMsg(m.SendMsgInfo.CheckMod, m.SendMsgInfo.Topic, m.SendMsgInfo.TableName, ms) {
//				errorWriteInfo(fmt.Sprintf("(%d) %v %s Sending data check topic %v message failed", logSeq, callFuncInfo, event, m.SendMsgInfo.Topic))
//				return false
//			}
//		}
//	}
//}
//	return false
//}

// 结束标记管道
//func expensiveOperationCount(wg *sync.WaitGroup, message mq.Message, done chan struct{}, sum int64, logSeq int64) {
//	var (
//		tableName = string(message.Body)
//		plan      = message.Properties["plan"].(map[string]*SchedulePlan2)[tableName]
//		event     = "[expensiveOperation]"
//		vlog      string
//	)
//	vlog = fmt.Sprintf("(%d) %v Start monitoring table %v.%v tasks and when to start the exit mechanism.", logSeq, event, plan.Object.Schema, plan.Object.Table)
//	debugWriteInfo(vlog)
//	defer func() {
//		done <- struct{}{}
//		wg.Done()
//	}()
//	ticker := time.NewTicker(global.MonitorInterval * time.Second)
//	for {
//		select {
//		case <-ticker.C:
//			if e := atomic.LoadInt64(plan.Status.ErrEnd); e < 0 {
//				return
//			}
//			if atomic.LoadInt64(plan.mqProductCustomerMonitor.MqStartTiming) == sum {
//				if atomic.LoadInt64(plan.mqProductCustomerMonitor.MqLevelProductSend) == atomic.LoadInt64(plan.mqProductCustomerMonitor.MqLevelCustomerReceive) {
//					vlog = fmt.Sprintf("(%d) %v Table %v.%v starts to start the exit task detection mechanism.", logSeq, event, plan.Object.Schema, plan.Object.Table)
//					debugWriteInfo(vlog)
//					return
//				}
//			}
//		}
//	}
//}

// 结束标记管道
//func expensiveOperation(wg *sync.WaitGroup, message mq.Message, done chan struct{}) {
//	var (
//		plan   = message.Properties["plan"].(*SchedulePlan2)
//		logSeq = plan.Object.ObjectLogSeq
//		event  = "[expensiveOperation]"
//	)
//	debugWriteInfo(fmt.Sprintf("(%d) %v Start monitoring table %v.%v tasks and when to start the exit mechanism.", logSeq, event, plan.Object.Schema, plan.Object.Table))
//	defer func() {
//		done <- struct{}{}
//		wg.Done()
//	}()
//	ticker := time.NewTicker(global.MonitorInterval * time.Second)
//	for {
//		select {
//		case <-ticker.C:
//			if abnormalErrorQuit(plan) {
//				return
//			}
//			//fmt.Println("-----ggg:", plan.Object.Schema, plan.Object.Table, atomic.LoadInt64(plan.mqProductCustomerMonitor.FirstMsg.ProductObject), atomic.LoadInt64(plan.mqProductCustomerMonitor.FirstMsg.CustomerObject))
//			//fmt.Println("-----fff:", atomic.LoadInt64(plan.mqProductCustomerMonitor.FirstMsg.ProductMsg), atomic.LoadInt64(plan.mqProductCustomerMonitor.FirstMsg.CustomerMsg))
//			//fmt.Println("-----kkk:", atomic.LoadInt64(plan.mqProductCustomerMonitor.FirstMsg.ProductSendMsgFinishMark))
//			if getFirstObjectProductCustomerEq(plan) {
//				//fmt.Println("-------jjj:", getFirstMsgProductCustomerEq(plan), "||", getFirstMsgFinishStatus(plan))
//				if getFirstMsgProductCustomerEq(plan) && getFirstMsgFinishStatus(plan) {
//					debugWriteInfo(fmt.Sprintf("(%d) %v Table %v.%v starts to start the exit task detection mechanism.", logSeq, event, plan.Object.Schema, plan.Object.Table))
//					return
//				}
//			}
//		}
//	}
//}

//// 每个表的mq消息状态监听，结束时发送状态给bar
//func mqMissTableStatusInterruptMonitor(done chan struct{}, wg *sync.WaitGroup, message mq.Message, logSeq int64) {
//	var (
//		tableName = string(message.Body)
//		event     = "[mqMissTableStatusInterruptMonitor]"
//		vlog      string
//		plan      = message.Properties["plan"].(map[string]*SchedulePlan2)[tableName]
//	)
//	defer func() {
//		wg.Done()
//	}()
//	vlog = fmt.Sprintf("(%d) %v Start monitoring the mq queue status of table %v.%v. The default timeout is 360 days.", logSeq, event, plan.Object.Schema, plan.Object.Table)
//	debugWriteInfo(vlog)
//	for {
//		select {
//		case d, ok := <-plan.mqProductCustomerMonitor.ProductMessage:
//			if !ok {
//				if e := atomic.LoadInt64(plan.Status.ErrEnd); e < 0 {
//					vlog = fmt.Sprintf("(%d) %v An error occurs during the execution of table %v.%v, and error handling is performed.", logSeq, event, plan.Object.Schema, plan.Object.Table)
//					infoWriteInfo(vlog)
//					if err := xlsWriterTerminalData(plan, logSeq); err != nil {
//						PlanContext.TaskBarSubsStatusSwap(plan.subTaskInfo, "error")
//						return
//					}
//					PlanContext.TaskBarSubsStatusSwap(plan.subTaskInfo, "error")
//					return
//				}
//				if *plan.mqProductCustomerMonitor.MqProductSeq == *plan.mqProductCustomerMonitor.MqCustomerSeq {
//					vlog = fmt.Sprintf("(%d) %v It is detected that the message queue production quantity and consumption quantity in table %v.%v are equal.", logSeq, event, plan.Object.Schema, plan.Object.Table)
//					debugWriteInfo(vlog)
//					if *plan.statistics.AbTotal > 0 {
//						atomic.SwapInt64(plan.statistics.AbTotal, 0)
//						sendMsg(getCheckMod(), RowsMissMd5ValueUniq, fmt.Sprintf("%v.%v", plan.Object.Schema, plan.Object.Table), map[string]any{"logSeq": logSeq, "segmentSql": map[string]string{}, "plan": message.Properties["plan"], "subTask": plan.subTaskInfo, "topic": string(RowsMissMd5ValueUniq)})
//					} else {
//						st := getTableName(plan.Object.Schema, plan.Object.Table)
//						if err := writeBreakPointData(GlobalPConfigs.FilePoint.BreakPointFin, []*string{&st}, logSeq); err != nil {
//							PlanContext.TaskBarSubsStatusSwap(plan.subTaskInfo, "finish")
//							return
//						}
//						if err := xlsWriterTerminalData(plan, logSeq); err != nil {
//							PlanContext.TaskBarSubsStatusSwap(plan.subTaskInfo, "finish")
//							return
//						}
//						PlanContext.TaskBarSubsStatusSwap(plan.subTaskInfo, "finish")
//					}
//					vlog = fmt.Sprintf("(%v) %v Table %v.%v data migration is completed.", logSeq, event, plan.Object.Schema, plan.Object.Table)
//					infoWriteInfo(vlog)
//					return
//				}
//
//			} else {
//				for _, v := range d {
//					atomic.AddInt64(plan.mqProductCustomerMonitor.MqProductSeq, int64(len(v)))
//				}
//			}
//		case <-done:
//			close(plan.mqProductCustomerMonitor.ProductMessage)
//			vlog = fmt.Sprintf("(%d) %v It is detected that there is no update in the message queue of table %v.%v within 5 seconds, and the queue writing is closed.", logSeq, event, plan.Object.Schema, plan.Object.Table)
//			debugWriteInfo(vlog)
//		}
//	}
//}

// 每个表的mq消息状态监听，结束时发送状态给bar
//func mqTableStatusMonitor(done, done1 chan struct{}, wg *sync.WaitGroup, message mq.Message) {
//	var (
//		plan   = message.Properties["plan"].(*SchedulePlan2)
//		logSeq = plan.Object.ObjectLogSeq
//		event  = "[mqTableStatusMonitor]"
//		Quit   bool
//	)
//	defer func() {
//		done1 <- struct{}{}
//		//terminalPods <- plan.TPod
//		newTer.Write(plan.TPod)
//		wg.Done()
//	}()
//	debugWriteInfo(fmt.Sprintf("(%d) %v Start monitoring the mq queue status of table %v.%v. The default timeout is 360 days.", logSeq, event, plan.Object.Schema, plan.Object.Table))
//	for {
//		select {
//		case <-time.NewTicker(1 * time.Second).C:
//			if !Quit {
//				continue
//			}
//			debugWriteInfo(fmt.Sprintf("(%d) %v It is detected that there is no update in the message queue of table %v.%v within 5 seconds, and the queue writing is closed.", logSeq, event, plan.Object.Schema, plan.Object.Table))
//			if abnormalErrorQuit(plan) {
//				mqStatusMonitorAbnormalQuit(plan, logSeq)
//				return
//			}
//			if getSecondMsgProductCustomerEq(plan) && getSecondMsgFinishStatus(plan) {
//				mqStatusMonitorNormalQuit(plan, logSeq)
//				return
//			}
//		case <-done:
//			Quit = true
//		}
//	}
//}

//func (t CheckSchemaMetaData) objectMetaInit1(seq int, v arg.InputTableAttributes, logSeq int64) (res *SchemaInitContext) {
//	var (
//		event = "[objectMetaInit]"
//		v1    = initMetaD(logSeq)
//	)
//	res = &SchemaInitContext{
//		Schema: v.Schema, Table: v.Table, MSchema: v.MSchema,
//		MTable: v.MTable, Seq: seq, ObjectLogSeq: logSeq,
//	}
//	res.TPod = t.podInit(seq, v)
//	if atomic.LoadInt64(v.TableStatus) == -1 {
//		res.TPod.TableMiss = true
//		return
//	}
//	for _, p := range []string{"struct", "index", "part"} {
//		fun, ok := getObjectMetaFunc(v1)[p]
//		if !ok {
//			return
//		}
//		args := []reflect.Value{
//			reflect.ValueOf(metaD.ObjectMetaInput{
//				TableAttributes: v,
//				TIndex:          res.IndexData,
//			}),
//		}
//		for y, x := range fun.Call(args) {
//			switch y {
//			case 0:
//				switch p {
//				case "struct":
//					res.TColumnMeta = x.Interface().(global.TableMetaGather)
//				case "index":
//					res.IndexData = x.Interface().(metaD.IndexChooseResult)
//				case "part":
//					res.Meta.Part = x.Interface().(global.PartMetaMarge)
//				}
//				res.TPod.IndexType = res.IndexData.IndexType
//				res.TPod.IndexCol = strings.Join(res.IndexData.IndexColumn, ",")
//			case 1:
//				if x.Interface() != nil {
//					errorWriteInfo(fmt.Sprintf("(%v) %v An error occurred in obtaining the table %v of table %v.%v and will be skipped.", logSeq, event, p, v.Schema, v.Table))
//					return
//				}
//			}
//		}
//	}
//	return
//}
